package com.group1.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.group1.realtime.app.BaseAppV1;
import com.group1.realtime.bean.BroadCastAccTimeBean;
import com.group1.realtime.common.Constant;
import com.group1.realtime.function.DimAsyncFunction;
import com.group1.realtime.util.AtguiguUtil;
import com.group1.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;


import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class Dws_BroadCast_AccTime extends BaseAppV1 {

    public static void main(String[] args) {
        new Dws_BroadCast_AccTime().init(
                3068,
                2,
                "Dws_BroadCast_AccTime",
                Constant.TOPIC_DWD_TRAFFIC_APPVIDEO
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env,
                          DataStreamSource<String> stream) {


        SingleOutputStreamOperator<BroadCastAccTimeBean> beanStream = stream
                .map(JSON::parseObject)
                .map(value -> {

                    JSONObject data = value.getJSONObject("appVideo");

                    return BroadCastAccTimeBean.builder()
                            .video_id(data.getString("video_id"))
                            .acc_time(data.getLong("play_sec"))
                            .ts(value.getLong("ts"))
                            .build();
                });

        SingleOutputStreamOperator<BroadCastAccTimeBean> streamWithDim = AsyncDataStream.unorderedWait(
                beanStream,
                new DimAsyncFunction<BroadCastAccTimeBean>() {

                    @Override
                    public String getTable() {
                        return "dim_video_info";
                    }

                    @Override
                    public String getId(BroadCastAccTimeBean input) {
                        return input.getVideo_id();
                    }

                    @Override
                    public void addDim(BroadCastAccTimeBean input,
                                       JSONObject dim) {
                        input.setChapter_id(dim.getString("CHAPTER_ID"));
                    }
                },
                60,   // 异步处理超时时间
                TimeUnit.SECONDS  // 超时赶时间的单位
        );

        streamWithDim
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<BroadCastAccTimeBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                                .withIdleness(Duration.ofSeconds(5))
                )
                .keyBy(BroadCastAccTimeBean::getChapter_id)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<BroadCastAccTimeBean>() {
                            @Override
                            public BroadCastAccTimeBean reduce(BroadCastAccTimeBean value1,
                                                               BroadCastAccTimeBean value2) throws Exception {
                                value1.setAcc_time(value1.getAcc_time() + value2.getAcc_time());
                                return value1;
                            }
                        },
                        new ProcessWindowFunction<BroadCastAccTimeBean, BroadCastAccTimeBean, String, TimeWindow>() {
                            @Override
                            public void process(String key,
                                                Context ctx,
                                                Iterable<BroadCastAccTimeBean> elements,
                                                Collector<BroadCastAccTimeBean> out) throws Exception {

                                BroadCastAccTimeBean bean = elements.iterator().next();

                                bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));

                                bean.setTs(ctx.currentProcessingTime());

                                out.collect(bean);

                            }
                        })
                //.print();
                .addSink(FlinkSinkUtil.getClickHouseSink("dws_broadcast_acc_time", BroadCastAccTimeBean.class));
    }
}
